/*
 * Copyright 2007 Matt Jensen
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 */
package org.jtell.internal;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jtell.EventSink;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * <p>
 * <code>ConcurrentEventSinkAffinityMultiMap</code> is an implementation of the {@link EventSinkAffinityMultiMap}
 * interface which is safe for concurrent access across multiple threads. All additions, removals, and queries are
 * handled in a thread safe manner, using test-and-set, copy-on-write semantics. This implementation is most efficient
 * when used in an environment where queries are frequent but additions and removals are rare.
 * </p>
 * <p>
 * <strong>Thread Safety</strong><br/>
 * Instances of this class are safe for multithreaded access. 
 * </p>
 */
public class ConcurrentEventSinkAffinityMultiMap implements EventSinkAffinityMultiMap
{
    private static Log LOG = LogFactory.getLog(ConcurrentEventSinkAffinityMultiMap.class);

    /**
     * <p>
     * Map of fully qualified event and/or source class names to the event sinks which have been associated with those
     * classes. Since any class in this map may be used as a source class, an event class, or both, each is associated
     * with an {@link AffinityEntry} which ties in those sinks which are associated with it as an event class or a
     * source class.
     * </p>
     */
    private final ConcurrentMap<String, AffinityEntry> m_affinity = new ConcurrentHashMap<String, AffinityEntry>();

    /**
     * <p>
     * Construct a {@link ConcurrentEventSinkAffinityMultiMap} instance.
     * </p>
     */
    public ConcurrentEventSinkAffinityMultiMap()
    {
        super();
    }

    /* inherit javadoc */
    public void clear()
    {
        m_affinity.clear();
    }

    /* inherit javadoc */
    public Set<EventSink> getAllForEventClass(final String eventClassName)
    {
        // Iterate over superclass names.
        Set<EventSink> result = Empties.EMPTY_EVENT_SINKS;
        for (final String className : ReflectionUtils.getAllSuperclassNames(eventClassName))
        {
            // Get the affinity entry associated with the next class name.
            final AffinityEntry affinityEntry = m_affinity.get(className);
            if (null != affinityEntry)
            {
                // Get the event class associations.
                final Set<EventSink> eventClassEventSinks = affinityEntry.getEventClassEventSinks();
                if (!eventClassEventSinks.isEmpty())
                {
                    // Allocate a new set on first non-empty event class association set.
                    if (Empties.EMPTY_EVENT_SINKS == result)
                    {
                        result = new HashSet<EventSink>();
                    }

                    // Add the event class event sinks for this class.
                    result.addAll(eventClassEventSinks);
                }
            }
        }
        if (LOG.isDebugEnabled())
        {
            LOG.debug(String.format("Returning %d event sinks for event class [%s].", result.size(), eventClassName));
        }
        return result;
    }

    /* inherit javadoc */
    public Set<EventSink> getAllForSourceClass(final String sourceClassName)
    {
        // Iterate over superclass names.
        Set<EventSink> result = Empties.EMPTY_EVENT_SINKS;
        for (final String className : ReflectionUtils.getAllSuperclassNames(sourceClassName))
        {
            // Get the affinity entry associated with the next class name.
            final AffinityEntry affinityEntry = m_affinity.get(className);
            if (null != affinityEntry)
            {
                // Get the source class associations.
                final Set<EventSink> sourceClassEventSinks = affinityEntry.getSourceClassEventSinks();
                if (!sourceClassEventSinks.isEmpty())
                {
                    // Allocate a new set on first non-empty source class association set.
                    if (Empties.EMPTY_EVENT_SINKS == result)
                    {
                        result = new HashSet<EventSink>();
                    }

                    // Add the source class event sinks for this class.
                    result.addAll(sourceClassEventSinks);
                }
            }
        }
        if (LOG.isDebugEnabled())
        {
            LOG.debug(String.format("Returning %d event sinks for source class [%s].", result.size(), sourceClassName));
        }
        return result;
    }

    /* inherit javadoc */
    public void putForEventClass(final String eventClassName, final EventSink eventSink)
    {
        // The update might require multiple tries if we encounter contention.
        while (true)
        {
            // Get existing affinity entry, if any.
            final AffinityEntry affinityEntry = m_affinity.get(eventClassName);
            if (null == affinityEntry)
            {
                // Found no existing entry; create an initial entry and attempt to store it.
                final Set<EventSink> newEventClassEventSinks = Collections.singleton(eventSink);
                final AffinityEntry newAffinityEntry = new AffinityEntry(newEventClassEventSinks,
                        Empties.EMPTY_EVENT_SINKS);
                if (null == m_affinity.putIfAbsent(eventClassName, newAffinityEntry))
                {
                    // We were able to add the entry; break.
                    break;
                }
            }
            else
            {
                // Found an existing entry. Get current event sink set; we will add only if this sink is not already
                // present.
                final Set<EventSink> eventClassEventSinks = affinityEntry.getEventClassEventSinks();
                if (eventClassEventSinks.contains(eventSink))
                {
                    // The event sink is already in the set; break.
                    break;
                }

                // Create a new affinity entry with the addition of the new event class event sink, then attempt to
                // replace the set.
                final Set<EventSink> newEventClassEventSinks;
                if (eventClassEventSinks.isEmpty())
                {
                    newEventClassEventSinks = Collections.singleton(eventSink);
                }
                else
                {
                    newEventClassEventSinks = new HashSet<EventSink>(eventClassEventSinks.size() + 1);
                    newEventClassEventSinks.addAll(eventClassEventSinks);
                    newEventClassEventSinks.add(eventSink);
                }
                final AffinityEntry newAffinityEntry = new AffinityEntry(newEventClassEventSinks,
                        affinityEntry.getSourceClassEventSinks());
                if (m_affinity.replace(eventClassName, affinityEntry, newAffinityEntry))
                {
                    // We were able to replace the entry; break.
                    break;
                }
            }
        }
    }

    /* inherit javadoc */
    public void putForSourceClass(final String sourceClassName, final EventSink eventSink)
    {
        // The update might require multiple tries if we encounter contention.
        while (true)
        {
            // Get existing affinity entry, if any.
            final AffinityEntry affinityEntry = m_affinity.get(sourceClassName);
            if (null == affinityEntry)
            {
                // Found no existing entry; create an initial entry and attempt to store it.
                final Set<EventSink> newSourceClassEventSinks = Collections.singleton(eventSink);
                final AffinityEntry newAffinityEntry = new AffinityEntry(Empties.EMPTY_EVENT_SINKS,
                        newSourceClassEventSinks);
                if (null == m_affinity.putIfAbsent(sourceClassName, newAffinityEntry))
                {
                    // We were able to add the entry; break.
                    break;
                }
            }
            else
            {
                // Found an existing entry. Get current event sink set; we will add only if this sink is not already
                // present.
                final Set<EventSink> sourceClassEventSinks = affinityEntry.getSourceClassEventSinks();
                if (sourceClassEventSinks.contains(eventSink))
                {
                    // The event sink is already in the set; break.
                    break;
                }

                // Create a new affinity entry with the addition of the new source class event sink, then attempt to
                // replace the set.
                final Set<EventSink> newSourceClassEventSinks;
                if (sourceClassEventSinks.isEmpty())
                {
                    newSourceClassEventSinks = Collections.singleton(eventSink);
                }
                else
                {
                    newSourceClassEventSinks = new HashSet<EventSink>(sourceClassEventSinks.size() + 1);
                    newSourceClassEventSinks.addAll(sourceClassEventSinks);
                    newSourceClassEventSinks.add(eventSink);
                }
                final AffinityEntry newAffinityEntry = new AffinityEntry(affinityEntry.getEventClassEventSinks(),
                        newSourceClassEventSinks);
                if (m_affinity.replace(sourceClassName, affinityEntry, newAffinityEntry))
                {
                    // We were able to replace the entry; break.
                    break;
                }
            }
        }
    }

    /* inherit javadoc */
    public void removeMatching(final Predicate<EventSink> predicate)
    {
        // Iterate over entries in the affinity map.
        for (final String className : m_affinity.keySet())
        {
            // The class sets must be updated, which may require multiple tries if we encounter contention.
            while (true)
            {
                // First check for any matches in the event and source class sets.
                final AffinityEntry affinityEntry = m_affinity.get(className);
                if (null == affinityEntry)
                {
                    // The entry under this key has been removed; break.
                    break;
                }
                boolean foundEventClassMatch = false;
                final Set<EventSink> eventClassEventSinks = affinityEntry.getEventClassEventSinks();
                for (final EventSink eventSink : eventClassEventSinks)
                {
                    if (predicate.evaluate(eventSink))
                    {
                        foundEventClassMatch = true;
                        break;
                    }
                }
                boolean foundSourceClassMatch = false;
                final Set<EventSink> sourceClassEventSinks = affinityEntry.getSourceClassEventSinks();
                for (final EventSink eventSink : sourceClassEventSinks)
                {
                    if (predicate.evaluate(eventSink))
                    {
                        foundSourceClassMatch = true;
                        break;
                    }
                }

                // Reallocate the sets where necessary.
                if (!foundEventClassMatch && !foundSourceClassMatch)
                {
                    // No matches exist; break.
                    break;
                }
                else
                {
                    // Remove matching entries from the event class set if necessary.
                    final Set<EventSink> newEventClassEventSinks;
                    if (!foundEventClassMatch)
                    {
                        newEventClassEventSinks = eventClassEventSinks;
                    }
                    else
                    {
                        // Copy the event class set, then remove the matches.
                        newEventClassEventSinks = new HashSet<EventSink>(eventClassEventSinks);
                        for (final Iterator<EventSink> itEventSink = newEventClassEventSinks.iterator();
                             itEventSink.hasNext();)
                        {
                            if (predicate.evaluate(itEventSink.next()))
                            {
                                itEventSink.remove();
                            }
                        }
                    }

                    // Remove matching entries from the source class set if necessary.
                    final Set<EventSink> newSourceClassEventSinks;
                    if (!foundSourceClassMatch)
                    {
                        newSourceClassEventSinks = sourceClassEventSinks;
                    }
                    else
                    {
                        // Copy the source class set, then remove the matches.
                        newSourceClassEventSinks = new HashSet<EventSink>(sourceClassEventSinks);
                        for (final Iterator<EventSink> itEventSink = newSourceClassEventSinks.iterator();
                             itEventSink.hasNext();)
                        {
                            if (predicate.evaluate(itEventSink.next()))
                            {
                                itEventSink.remove();
                            }
                        }
                    }

                    // Update the affinity map.
                    if (newEventClassEventSinks.isEmpty() && newSourceClassEventSinks.isEmpty())
                    {
                        // Both sets are empty, so we can simply remove the map entry.
                        if (m_affinity.remove(className, affinityEntry))
                        {
                            // We were able to remove the entry; break.
                            break;
                        }
                    }
                    else
                    {
                        // At least one set still holds a value, so replace the map entry.
                        final AffinityEntry newAffinityEntry = new AffinityEntry(newEventClassEventSinks,
                                newSourceClassEventSinks);
                        if (m_affinity.replace(className, affinityEntry, newAffinityEntry))
                        {
                            // We were able to replace the entry; break.
                            break;
                        }
                    }
                }
            }
        }
    }

    /**
     * <p>
     * <code>AffinityEntry</code> holds sets of {@link EventSink} objects which have been associated with an event or
     * source class name in the affinity map.
     * </p>
     */
    private static class AffinityEntry
    {
        /**
         * <p>
         * Event sinks for which this is the event class.
         * </p>
         */
        private final Set<EventSink> m_eventClassEventSinks;

        /**
         * <p>
         * Event sinks for which this is the source class.
         * </p>
         */
        private final Set<EventSink> m_sourceClassEventSinks;

        /**
         * <p>
         * Construct an {@link AffinityEntry} instance.
         * </p>
         *
         * @param eventClassEventSinks  event sinks for which this is the event class.
         * @param sourceClassEventSinks event sinks for which this is the source class.
         */
        private AffinityEntry(final Set<EventSink> eventClassEventSinks, final Set<EventSink> sourceClassEventSinks)
        {
            super();
            Guard.notNull("eventClassEventSinks", eventClassEventSinks);
            Guard.notNull("sourceClassEventSinks", sourceClassEventSinks);

            m_eventClassEventSinks = eventClassEventSinks;
            m_sourceClassEventSinks = sourceClassEventSinks;
        }

        /**
         * <p>
         * Get the event sinks associated with the event class.
         * </p>
         *
         * @return {@link Set} of {@link EventSink} instances.
         */
        public Set<EventSink> getEventClassEventSinks()
        {
            return m_eventClassEventSinks;
        }

        /**
         * <p>
         * Get the event sinks associated with the source class.
         * </p>
         *
         * @return {@link Set} of {@link EventSink} instances.
         */
        public Set<EventSink> getSourceClassEventSinks()
        {
            return m_sourceClassEventSinks;
        }
    }
}
